#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <sys/time.h>
#include <signal.h>

#include "utils/msocket.h"
#include "utils/log.h"
#include "utils/argopt.h"
#include "utils/jsoncx.h"
#include "utils/base.h"
#include "comm/mqtt.h"

static inline int ipv4_check(const char *ip)
{
    if (ip == NULL || ip[0] == '0' || ip[0] == '\0')
    {
        return -1;
    }

    for (int i = 0, count = 0; i < (int)strlen(ip); i++)
    {
        if ((ip[i] != '.') && (ip[i] < '0' || ip[i] > '9'))
        {
            return -1;
        }
        if (ip[i] == '.')
        {
            count++;
            if (count > 3)
            {
                return -1;
            }
        }
    }

    int ip_num[4] = {-1, -1, -1, -1};
    char ip_s[4][4];
    memset(ip_s, 0, sizeof(char[4]) * 4);

    sscanf(ip, "%[^.].%[^.].%[^.].%[^ ]", ip_s[0], ip_s[1], ip_s[2], ip_s[3]);
    sscanf(ip_s[0], "%d", &ip_num[0]);
    sscanf(ip_s[1], "%d", &ip_num[1]);
    sscanf(ip_s[2], "%d", &ip_num[2]);
    sscanf(ip_s[3], "%d", &ip_num[3]);

    for (int i = 0; i < 4; i++)
    {
        if (strlen(ip_s[i]) == 0 || (ip_s[i][0] == '0' && ip_s[i][1] != '\0') || ip_num[i] < 0 || ip_num[i] > 255)
        {
            return -1;
        }
    }

    return 0;
}



char ipv4_buf[16] = { "0.0.0.0" };
int ip_set(argopt_ctl_t *ctl, argopt_t *self, const char *arg)
{
	if (ipv4_check(arg))
	{
		log_error("ipv4_check(%s) fail now use 0.0.0.0", arg);
		memset(ipv4_buf, 0, sizeof(ipv4_buf));
	}
	else
	{
		strncpy(ipv4_buf, arg, sizeof(ipv4_buf));
	}
	return 0;
}


uint16_t port = 1883;
int port_set(argopt_ctl_t *ctl, argopt_t *self, const char *arg)
{
	int p = atoi(arg);
	if (p <= 1024 || p > 65536)
	{
		log_error("port fail now use 1883");
	}
	else
	{
		port = p;
	}
	return 0;
}
char client_id_default[16] = {'s','m','q','t','t','_'};
int client_id_set(argopt_ctl_t *ctl, argopt_t *self, const char *arg)
{
	self->data = strdup(arg);
	return 0;
}

int username_set(argopt_ctl_t *ctl, argopt_t *self, const char *arg)
{
	self->data = strdup(arg);
	return 0;
}

int password_set(argopt_ctl_t *ctl, argopt_t *self, const char *arg)
{
	self->data = strdup(arg);
	return 0;
}

void chars_free(argopt_t *self)
{
	if (self->data!=NULL)
		free(self->data);
}

void subfile_free(argopt_t *self)
{
	jsoncx_delete((jsoncx *)self->data);
}

int subfile_set(argopt_ctl_t *ctl, argopt_t *self, const char *arg)
{
	struct stat st;
	if (stat(arg, &st))
		return -1;
	char *buf = malloc(st.st_size + 1);
	if (buf == NULL)
		return -1;
	FILE *fp = fopen(arg, "r");
	if (fp == NULL)
		return -1;
	size_t n = fread(buf, sizeof(char), st.st_size, fp);
	if (n != st.st_size)
	{
		printf("n(%ld) != st.st_size(%ld)\n", n, st.st_size);
	}
	else
	{
		jsoncx *root = jsoncx_parse(buf);
		// int len = jsoncx_getarraysize(root);
		// for (size_t i = 0; i < len; i++)
		// {
		// 	jsoncx* item = jsoncx_getarrayitem(root,i);
		// 	jsoncx *topic = jsoncx_getobjectitem(item, "topic");
		// 	jsoncx *qos = jsoncx_getobjectitem(item, "qos");
		// 	log_debug("[%ld]topic:%s qos:%d", i, topic->valuestring,qos->valueint);
		// }
		argopt_setdata(self,root);
	}
	free(buf);
	fclose(fp);
	return 0;
}
u16 alive_sec = 100;
int alive_set(argopt_ctl_t *ctl, argopt_t *self, const char *arg)
{
	u16 sec = (u16) atoi(arg);
	alive_sec = sec != 0 ? sec : alive_sec;
	return 0;
}
int help(argopt_ctl_t *ctl, argopt_t *self, const char *arg)
{
	printf("Usage:\n"
		"\tI set server ip [0.0.0.0]\n"
		"\tP set server port [1883]\n"
		"\tc set client_id\n"
		"\tu set username\n"
		"\tp set password\n"
		"\tf set subfile\n"
	);
	exit(0);
}
static u8 do_sock_loop = 1;
void stop_sock_loop(int sig)
{
	log_info("ctrl + c signal!");
	do_sock_loop = 0;
}

// gcc ./sub.c -I../../ -I../ ../comm/*.c ../../utils/log.c ../../utils/argopt.c ../../utils/msocket.c ../../utils/jsoncx.c ../../utils/base.c -o sub -lm
// arm-rockchip830-linux-uclibcgnueabihf-gcc 
int main(int argc, char *argv[])
{
	signal(SIGINT, stop_sock_loop);
	argopt_ctl_t ctl = argopt_ctl();
	argopt_t arr[] = {
		argopt('I',0,ip_set,NULL),
		argopt('P',0,port_set,NULL),
		argopt('c',ARGOPT_DATAENABLE,client_id_set,chars_free),
		argopt('u',ARGOPT_DATAENABLE,username_set,chars_free),
		argopt('p',ARGOPT_DATAENABLE,password_set,chars_free),
		argopt('f',ARGOPT_DATAENABLE,subfile_set,subfile_free),
		argopt('a',ARGOPT_DATAENABLE,alive_set,NULL),
		argopt('h',0,help,NULL),
	};
	int err = argopt_ctl_init(&ctl, arr, argopt_arr_len(arr), argc, argv);
	if (err)
	{
		log_error("argopt_ctl_init return %d", err);
	}
	else
	{
		int step = 0;
		socket_t sock = socket_invalid;
		do
		{
			jsoncx *root = argopt_ctl_getdata(&ctl, 'f');

			if (root == NULL)
			{
				log_error("not set subfile -f to set this");
				break;
			}
			int sub_len = jsoncx_getarraysize(root);
			u8 **topics = (u8 **) malloc(sizeof(char *) * sub_len);
			u8 *qoss = (u8 *) malloc(sizeof(u8) * sub_len);
			if (topics == NULL || qoss == NULL)
			{
				log_error("OOM can't malloc topics or qoss");
				break;
			}
			for (size_t i = 0; i < sub_len; i++)
			{
				jsoncx *item = jsoncx_getarrayitem(root, i);
				jsoncx *topic = jsoncx_getobjectitem(item, "topic");
				jsoncx *qos = jsoncx_getobjectitem(item, "qos");
				log_debug("[%ld]topic:%s qos:%d", i, topic->valuestring, qos->valueint);
				topics[i] = topic->valuestring;
				qoss[i] = qos->valueint;
			}
			
			int ret = socket_init();
			if (ret)
			{
				log_error("socket_init fail:%d!", ret);
				break;
			}
			step++;
			sock = socket_tcp();
			if (sock == socket_invalid)
			{
				log_error("socket_tcp() fail:%d errstr:%s!", sock,strerror(socket_geterror()));
				break;
			}
			step++;
			ret = socket_connect_ipv4(sock, ipv4_buf, port);
			if (ret)
			{
				log_error("socket_connect_ipv4(%d,%s,%d) fail:%d errstr:%s!", sock, ipv4_buf, port, ret,strerror(socket_geterror()));
				break;
			}
			step++;
			u64 pack_len = 0;
			u16 seq = 1;
			long long last_ping_ms = now_ms();
			char *mqtt_client_id = (char *) argopt_ctl_getdata(&ctl, 'c');
			if (mqtt_client_id == NULL)
			{
				size_t client_id_default_len = strlen(client_id_default);
				get_rand_str(client_id_default + client_id_default_len, sizeof(client_id_default) - client_id_default_len);
				mqtt_client_id = client_id_default;
			}
			char *mqtt_username = (char *) argopt_ctl_getdata(&ctl, 'u');
			char *mqtt_password = (char *) argopt_ctl_getdata(&ctl, 'p');
			int mqtt_alive_sec = alive_sec;
			u8 *data = NULL;
			
			int mqtt_connect_step = 0;
			int mqtt_subscribe_step = 0;
			fd_set rfds, wfds;
			struct timeval timeout;
			u8 buf[2048] = { 0 };
			while (do_sock_loop)
			{
				FD_ZERO(&rfds);
				FD_ZERO(&wfds);
				
				FD_SET(sock, &rfds);
				FD_SET(sock, &wfds);
				timeout = (struct timeval){
					.tv_sec = 5,
					.tv_usec = 0,
				};
				int select_ret = socket_select(sock + 1, &rfds, &wfds, NULL, &timeout);
				if (select_ret <= 0)
				{
					if (select_ret)
					{
						log_error("socket_select() fail:%d errstr:%s", select_ret, strerror(socket_geterror()));
						break;
					}
					else
					{
						log_debug("socket_select() time out");
						continue;
					}
				}
				memset(buf, 0, sizeof(buf));
				
				if (FD_ISSET(sock, &rfds))// 可读
				{
					ret = socket_recv(sock, buf, 1, 0); // 先把固定报头读出来
					if (ret <= 0)
					{
						if (ret == 0)
							log_info("socket_recv() server socket close fail:%d errstr:%s", ret, strerror(socket_geterror()));
						else
							log_info("socket_recv() fail:%d errstr:%s", ret, strerror(socket_geterror()));
						break;
					}
					mqtt_pack_type type = mqtt_unknow;
					// 解析服务端发送的数据包
					mqtt_unpacket_type(buf[0], &type);
					log_debug("buf[0]:%x head type:%d", buf[0], type);
					switch (type)
					{
					case mqtt_connack:
						if (mqtt_connect_step == 1)
						{
							socket_recv(sock, buf + 1, mqtt_connack_packet_len - 1, 0);
							if (strncmp(mqtt_connack_packet, buf, mqtt_connack_packet_len) == 0)
							{
								log_debug("mqtt connack suf!");
								mqtt_connect_step = 2;
							}
						}
						else
						{
							if (mqtt_connect_step == 0)
								log_info("mqtt connect not send ! what the fuck send connack to me!");
							else if (mqtt_connect_step == 2)
								log_info("mqtt connect is finish! what the fuck one more send connack to me!");
						}
						break;
					case mqtt_suback:
						if (mqtt_subscribe_step == 1)
						{
							data = mqtt_suback_packet(qoss, sub_len, &pack_len, seq);
							socket_recv(sock, buf + 1, pack_len - 1, 0);
							if (strncmp(data, buf, pack_len) == 0)
							{
								log_debug("mqtt suback suf!");
								mqtt_subscribe_step = 2;
							}
							else // 可能有些qos等级与要求的qos等级不同
							{
								log_debug("mqtt suback some subscribe topic qos != suback topic qos");
								for (size_t i = 0; i < sub_len; i++)
								{
									if (qoss[i] != buf[pack_len - sub_len + i])
										log_warning("mqtt sever suback topics[%d]:%s cli qos:%d server qos:%d",
											i, topics[i], qoss[i], buf[pack_len - sub_len + i]);
								}
								mqtt_subscribe_step = 2;

							}
							free(data);
							seq++;
						}
						break;
					case mqtt_pingresp:
						socket_recv(sock, buf + 1, mqtt_pingresp_packet_len - 1, 0);
						if (strncmp(buf, mqtt_pingresp_packet, mqtt_pingresp_packet_len) == 0)
						{
							log_debug("pingresp suf!");
						}
						break;
					case mqtt_publish:
						// 先读取长度
						socket_recv(sock, buf + 1, 4, 0);
						u8 len_level = 0;
						u32 publish_len = mqtt_decode_remaining_len(buf + 1, &len_level);
						log_debug("len_level:%d publish_len:%d", len_level, publish_len);
						socket_recv(sock, buf + 1 + 4, publish_len - 4 + len_level, 0);
						mqtt_publish_data_t detail = {0};
						mqtt_unpacket_publish_data(buf[0], publish_len, buf + 1 + len_level, &detail);
						log_debug("detail{dup:%d,qos:%d,retain:%d,topic_len%d,topic:0x%x,seq:%d,data:0x%x}"
							, detail.dup, detail.qos, detail.retain, detail.topic_len, detail.topic, detail.seq, detail.data);
						/* code */
						u8 *topic = malloc(detail.topic_len + 1);
						if (topic != NULL)
						{
							memcpy(topic, detail.topic, detail.topic_len);
							log_debug("topic:%s", topic);
							free(topic);
						}
						log_debug("data:%s", detail.data);
						/* code */
						u8 send_buf[8] = { 0 };
						u8 recv_buf[8] = { 0 };
						if (detail.qos == 1)
						{
							mqtt_puback_packet(send_buf, detail.seq);
							if (socket_send(sock, send_buf, mqtt_puback_packet_len, 0)>= 0)
								log_debug("puback send suf!");
						}
						else if (detail.qos == 2)
						{
							mqtt_pubrec_packet(send_buf, detail.seq);
							if (socket_send(sock, send_buf, mqtt_pubrec_packet_len, 0) >= 0)
							{
								log_debug("pubrec send suf!");
								u8 tmp_buf[8] = { 0 };
								mqtt_pubrel_packet(recv_buf, detail.seq);
								socket_recv(sock, tmp_buf, mqtt_pubrel_packet_len, 0);
								if (strncmp(recv_buf, tmp_buf, mqtt_pubrel_packet_len) == 0)
								{
									log_debug("pubrel recv suf!");
									mqtt_pubcomp_packet(send_buf, detail.seq);
									if (socket_send(sock, send_buf, mqtt_pubcomp_packet_len, 0)>= 0)
										log_debug("pubcomp send suf!");	
								}
							}
							
						}
						seq++;
						break;
					default:
						break;
					}
				}
				
				if (FD_ISSET(sock, &wfds)) // 可写
				{
					if (mqtt_connect_step == 0)
					{
						data = mqtt_connect_packet(mqtt_client_id, mqtt_username, mqtt_password, mqtt_alive_sec, 1,&pack_len);
						ret = socket_send(sock, data, pack_len, 0);
						free(data);
						log_debug("mqtt connect send suf");
						mqtt_connect_step = 1;
					}
					if (mqtt_connect_step != 2) // 等到服务器发送connack后才允许走下去
						continue;
					if (mqtt_subscribe_step == 0)
					{
						data = mqtt_subscribe_packet(topics, qoss, sub_len, &pack_len, seq);
						ret = socket_send(sock, data, pack_len, 0);
						free(data);
						log_debug("mqtt subscribe send suf");
						mqtt_subscribe_step = 1;
					}
					// 心跳
					long long nowtime = now_ms();
					if (nowtime - last_ping_ms >= mqtt_alive_sec * 1000)
					{
						socket_send(sock, mqtt_pingreq_packet, mqtt_pingreq_packet_len, 0);
						last_ping_ms = nowtime;
					}
				}

			}
			data = mqtt_unsubscribe_packet(topics, sub_len, &pack_len, seq);
			if (socket_send(sock, data, pack_len, 0) >= 0)
			{
				memset(buf, 0, sizeof(buf));
				u8 unsuback_buf[8] = { 0 };
				mqtt_unsuback_packet(unsuback_buf, seq);
				socket_recv(sock, buf, mqtt_unsuback_packet_len, 0);
				if (strncmp(buf, unsuback_buf, mqtt_unsuback_packet_len) == 0)
				{
					log_info("unsub suf!");
				}
			}
			socket_send(sock, mqtt_disconnect_packet, mqtt_disconnect_packet_len, 0);
		} while (0);
		if (step > 2)
			socket_close(sock);
		if (step > 1)
			socket_cleanup();
	}

	argopt_ctl_deinit(&ctl);
	exit(0);
}